package com.dxf.bigdata.D05_spark_again.案例

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object 下单支付数统计 {

  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("app")
    sparkConf.set("spark.port.maxRetries", "100")
    val sc = new SparkContext(sparkConf)

    //取数
    val rdd: RDD[String] = sc.textFile("datas/user_visit_action.txt")

    //过滤 点击数量,品类Id 不是null的记录
    val clickFilterRdd: RDD[String] = rdd.filter(x => {
      val line: Array[String] = x.split("_")
      line(6) != null && line(6) != "-1"//line(6)  第6位品类Id
    })

    // 点击数
    val clickCountRdd: RDD[(String, Int)] = clickFilterRdd.map(
      x => {
        val line: Array[String] = x.split("_")
        (line(6), 1)
      }
    ).reduceByKey(_ + _)


    //下单数  品类ids  8
    val orderFilterRdd: RDD[String] = rdd.filter(x => {
      val line: Array[String] = x.split("_")
      line(8) != "null"
    })

    val orderCountRdd: RDD[(String, Int)] = orderFilterRdd.flatMap(x => {

      val line: Array[String] = x.split("_")
      line(8).split(",")

    }).map((_, 1)).reduceByKey(_ + _)


    // 支付数 品类ids 10

    val payFilterRdd: RDD[String] = rdd.filter(x => {
      val line: Array[String] = x.split("_")
      line(10) != "null"
    })

    val payCountRdd: RDD[(String, Int)] = payFilterRdd.flatMap(x => {
      val line: Array[String] = x.split("_")
      val ids: Array[String] = line(10).split(",")
      ids
    }).map((_, 1)).reduceByKey(_ + _)


    //品类id排序取前10 ,点击数优先,相同点击数的下单数优先,相同下单数的支付数优先


    println(clickCountRdd.cogroup(orderCountRdd, payCountRdd).mapValues {

      case (cIter, oIter, pIter) =>
        var c = 0
        if (cIter.iterator.hasNext) {
          c =  cIter.head
        }
        var o = 0
        if (oIter.iterator.hasNext) {
          o = cIter.head
        }
        var p = 0
        if (pIter.iterator.hasNext) {
          p =  cIter.head
        }
        (c, o, p)
    }.sortBy(_._2, false).take(10).mkString(","))
    sc.stop()

  }

}
